package com.leilei;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.util.Arrays;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/21 13:21
 */
public class Flink_Table_1_demo {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        DataStreamSource<GirlFriend> scFriendList = env.fromCollection(Arrays.asList(new GirlFriend("小丽", "A", 22),
                new GirlFriend("小丽于", "C", 24),
                new GirlFriend("小胡1", "D", 23),
                new GirlFriend("小丽于", "D", 17),
                new GirlFriend("小胡2", "D", 25)));
        Table scTable = tableEnv.fromDataStream(scFriendList,$("name").as("friendName"),
                $("cup").as("likeCup"),
                $("age"));
        Table table = scTable.select($("friendName"), $("likeCup"),$("age"))
                .filter($("age").isGreater(18))
                .filter($("friendName").like("%小%"));
        DataStream<Friend> girlFriendDataStream = tableEnv.toAppendStream(table, Friend.class);
        girlFriendDataStream.print();
        //4> Flink_Table_1_demo.Friend(friendName=小胡2, likeCup=D, age=25)
        //1> Flink_Table_1_demo.Friend(friendName=小丽, likeCup=A, age=22)
        //3> Flink_Table_1_demo.Friend(friendName=小胡1, likeCup=D, age=23)
        //2> Flink_Table_1_demo.Friend(friendName=小丽于, likeCup=C, age=24)
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Friend {
        private String friendName;
        private String likeCup;
        private Integer age;
    }
}
